package alfred.mdp.broker;

import alfred.mdp.Majordomo;
import alfred.mdp.messages.worker.Disconnect;
import alfred.mdp.messages.worker.Heartbeat;
import alfred.mdp.utils.Shutdowner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;

import static alfred.mdp.utils.Utils.asString;

/**
 */
public final class WorkerReference implements Runnable {
    private static final Logger LOG = LoggerFactory.getLogger(WorkerReference.class);
    private static final String DEFAULT_LOG_MESSAGE = "WorkerReference of worker {} and service {} - ";
    private final String serviceName;
    private final ByteBuffer workerId;
    private final Broker broker;

    private volatile int currentLiveness = Majordomo.HB_LIVENESS;

    public WorkerReference(Broker broker, String serviceName, ByteBuffer workerId) {
        this.broker = broker;
        this.serviceName = serviceName;
        this.workerId = workerId;
        Shutdowner.add(getShutdownGroup(), broker.getExecutor().submit(new WorkerReferenceDispatcher(broker, this)));
    }

    /**
     * called every interval to send heartbeat to worker.
     */
    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                currentLiveness--;
                Thread.sleep(Majordomo.HB_INTERVAL);
                if (LOG.isTraceEnabled()) {
                    LOG.trace(DEFAULT_LOG_MESSAGE + "send heartbeat", asString(workerId), serviceName);
                }
                broker.sendMessageToWorker(new Heartbeat(), workerId.array());

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();

            } catch (Exception e) {
                LOG.error("Error", e);

            }
        }
        Shutdowner.shutdownCompleted(getShutdownGroup(), getClass());
        LOG.debug(DEFAULT_LOG_MESSAGE + "shutdown completed", asString(workerId), serviceName);
    }

    public void onHeartbeat() {
        currentLiveness = Majordomo.HB_LIVENESS;
        if (LOG.isTraceEnabled()) {
            LOG.trace(DEFAULT_LOG_MESSAGE + "onHeartbeat - reset liveness", asString(workerId), serviceName);
        }
    }

    public boolean isAlive() {
        if (LOG.isTraceEnabled()) {
            LOG.trace(DEFAULT_LOG_MESSAGE + "isAlive {}", new Object[]{asString(workerId), serviceName, currentLiveness > 0});
        }
        return currentLiveness > 0;
    }

    public boolean shutdown(boolean sendDisconnect) {
        if (LOG.isTraceEnabled()) {
            LOG.trace(DEFAULT_LOG_MESSAGE + "shutdown", asString(workerId), serviceName);
        }
        if (sendDisconnect) {
            broker.sendMessageToWorker(new Disconnect(), workerId.array());
        }
        return Shutdowner.shutdown(getShutdownGroup());

    }

    public String getShutdownGroup() {
        return getClass().getName() + getWorkerId().asCharBuffer();
    }

    public ByteBuffer getWorkerId() {
        return workerId;
    }

    public String getServiceName() {
        return serviceName;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;

        WorkerReference that = (WorkerReference) o;

        if (workerId != null ? !workerId.equals(that.workerId) : that.workerId != null) return false;

        return true;
    }

    @Override
    public int hashCode() {
        return workerId != null ? workerId.hashCode() : 0;
    }
}
